package com.hzw.saas.service.storage.service;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

import com.hzw.saas.api.storage.ICatalogService;
import com.hzw.saas.api.storage.IResourceFileService;
import com.hzw.saas.api.storage.IRsyncService;
import com.hzw.saas.api.storage.bo.*;
import com.hzw.saas.api.storage.constants.StorageCacheConsts;
import com.hzw.saas.api.storage.enums.FileIntegrityEnum;
import com.hzw.saas.api.storage.enums.ResourceStatusEnum;
import com.hzw.saas.common.config.exception.DefaultSaasException;
import com.hzw.saas.common.config.query.LogicQuery;
import com.hzw.saas.common.config.util.AssertUtil;
import com.hzw.saas.common.config.util.RedisUtil;
import com.hzw.saas.common.storage.enums.ChecksumModeEnum;
import com.hzw.saas.common.storage.rsync.*;
import com.hzw.saas.common.storage.service.IChunkStorageService;
import com.hzw.saas.common.storage.service.IFileStorageService;
import com.hzw.saas.common.storage.util.ChecksumUtil;
import com.hzw.saas.common.storage.util.StorageHubUtil;
import com.hzw.saas.common.util.MapperUtil;
import com.hzw.saas.common.util.SnowFlakeUtil;
import com.hzw.saas.common.util.constants.CacheConsts;
import com.hzw.saas.common.util.enums.AssertEnum;
import com.hzw.saas.common.util.enums.ConflictModeEnum;
import com.hzw.saas.common.util.exception.UnavailableException;
import com.hzw.saas.service.storage.config.StorageConfig;
import com.hzw.saas.service.storage.util.StorageUtil;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.lang.Assert;
import cn.hutool.core.thread.ThreadUtil;
import cn.hutool.core.util.StrUtil;
import org.springframework.http.HttpStatus;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @author EZ09
 * @since 07/07/2021
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class RsyncServiceImpl implements IRsyncService {

    private final PlatformTransactionManager transactionManager;

    private final StorageConfig storageConfig;
    private final ICatalogService catalogService;
    private final IResourceFileService fileService;
    private final IFileStorageService fileStorageService;
    private final IChunkStorageService chunkStorageService;

    @Override
    public SyncVersionFileBO getVersionInfo(String fileId) {
        ResourceFileBO requestFile = fileService.getFile(fileId, null, null, AssertEnum.FOUND);
        ResourceFileBO latestFile = fileService.getLatestFile(requestFile.getVid(),
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), null, AssertEnum.FOUND);
        this.checkFileIntegrity(latestFile);
        return new SyncVersionFileBO(latestFile, requestFile);
    }

    @Override
    public ServerOriginFileBO listServerOriginBlocks(String fileId, Integer blockLength, Integer blockStep, Integer checksumMode) {
        // check params
        AssertUtil.assertThrow("blockLength必须大于0", Objects.isNull(blockLength) || blockLength <= 0);
        if (Objects.isNull(blockStep)) blockStep = blockLength;
        // check file status
        ResourceFileStorageBO fileInfo = fileService.getFileWithStorage(fileId,
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
        this.checkFileIntegrity(fileInfo);
        // 获取缓存
        List<OriginBlockBO> originBlockBOList = this.getBlockChecksumCache(fileId, blockLength, blockStep);
        if (CollectionUtil.isEmpty(originBlockBOList)) {
            // 无缓存时
            // get file data
            File fileData = fileService.getFileData(fileId, AssertEnum.FOUND);
            // get sync block
            OriginBlockBuilder builder = new OriginBlockBuilder(fileData, blockLength, blockStep);
            List<OriginBlock> originBlocks = builder.build(checksumMode);
            originBlockBOList = MapperUtil.nf().mapAsList(originBlocks, OriginBlockBO.class);
            // 异步执行分片校验和缓存
            this.asyncAddBlockChecksumCache(fileId, builder);
        }
        // result
        RsyncTaskBO rsyncTaskBO = this.addSyncTaskCache(fileId, blockLength, blockStep);
        ServerOriginFileBO result = new ServerOriginFileBO();
        result.setSyncId(rsyncTaskBO.getSyncId());
        result.setOriginBlocks(originBlockBOList);
        result.setStrongChecksum(fileInfo.getStrongChecksum());
        result.setBlockLength(blockLength);
        return result;
    }

    @Override
    public List<OriginBlockBO> listServerOriginBlocks2(String syncId, List<Long> blockPosList, Integer checksumMode) {
        if (CollectionUtil.isEmpty(blockPosList))
            return new ArrayList<>();
        RsyncTaskBO syncTaskCache = this.getSyncTaskCache(syncId);
        if (Objects.isNull(syncTaskCache)) {
            throw new UnavailableException("同步任务已失效或不存在");
        }
        // 获取缓存
        List<OriginBlockBO> blockChecksumCache = this.getBlockChecksumCache(syncTaskCache.getFileId(), syncTaskCache.getBlockLength(), syncTaskCache.getBlockStep());
        if (Objects.nonNull(blockChecksumCache) && blockChecksumCache.size() > 0) {
            List<OriginBlockBO> result = new ArrayList<>();
            for (Long blockPos : blockPosList) {
                OriginBlockBO resultEntity = blockChecksumCache.stream().filter(block -> Objects.equals(block.getBlockPos(), blockPos)).findAny().orElse(null);
                AssertUtil.assertThrow(String.format("当前同步任务不存在pos: %s", blockPos), Objects.isNull(resultEntity));
                result.add(resultEntity);
            }
            return result;
        } else {
            // 无缓存时
            // check file status
            ResourceFileStorageBO fileInfo = fileService.getFileWithStorage(syncTaskCache.getFileId(),
                LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
            this.checkFileIntegrity(fileInfo);
            // get file data
            File fileData = fileService.getFileData(syncTaskCache.getFileId(), AssertEnum.FOUND);
            List<OriginBlock> originBlocks;
            try {
                originBlocks = RsyncFileUtil.calcBlockChecksum(fileData, blockPosList, syncTaskCache.getBlockLength(), checksumMode);
            } catch (IOException e) {
                throw DefaultSaasException.build(e.getMessage(), e);
            }
            // 异步缓存
            // get sync block
            OriginBlockBuilder builder = new OriginBlockBuilder(fileData, syncTaskCache.getBlockLength(), syncTaskCache.getBlockStep());
            this.asyncAddBlockChecksumCache(syncTaskCache.getFileId(), builder);
            return MapperUtil.nf().mapAsList(originBlocks, OriginBlockBO.class);
        }
    }

    @Override
    public ResourceFileBO receiveClientDiffBlocks(ClientDiffFileBO clientDiffBlocks) {
        Assert.notNull(clientDiffBlocks, "差异块数据为null");
        Assert.isTrue(CollectionUtil.isNotEmpty(clientDiffBlocks.getDiffBlocks()), "差异块列表为空");
        RsyncTaskBO syncTaskCache = this.getSyncTaskCache(clientDiffBlocks.getSyncId());
        if (Objects.isNull(syncTaskCache)) {
            throw new UnavailableException("同步任务已失效或不存在");
        }
        String fileId = syncTaskCache.getFileId();
        // check file status
        ResourceFileStorageBO srcFileInfo = fileService.getFileWithStorage(fileId,
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
        this.checkFileIntegrity(srcFileInfo);
        // source file
        File srcFile = fileService.getFileData(fileId, AssertEnum.FOUND);
        // new file
        String newFileId = SnowFlakeUtil.nextIdStr();
        String destFilePath = catalogService.getFilePath(storageConfig.getPersonalSpaceCatalog(), newFileId);
        File destFile = fileStorageService.getFile(destFilePath);
        // 更新状态: 正在合并...
        Integer originFileIntegrity = srcFileInfo.getIntegrity();
        fileService.updateFileIntegrity(fileId, FileIntegrityEnum.MERGING.getCode());

        // 开启事务
        DefaultTransactionDefinition def = new DefaultTransactionDefinition();
        def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
        TransactionStatus status = transactionManager.getTransaction(def);
        try {
            // merge
            String blockSupPath = catalogService.getCacheSupPath(storageConfig.getPersonalSpaceCacheCatalog(), clientDiffBlocks.getSyncId());
            RsyncFileUtil.mergeFile(srcFile, destFile, blockSupPath, MapperUtil.nf().mapAsList(clientDiffBlocks.getDiffBlocks(), DiffBlock.class));
            // 校验新文件
            // long weakChecksum = ChecksumUtil.adler32(destFile);
            String strongChecksum = ChecksumUtil.md5Hex(destFile);
            this.assertChecksum(clientDiffBlocks.getStrongChecksum(), strongChecksum);
            // 恢复旧文件状态
            // fileService.updateFileIntegrity(fileId, originFileIntegrity);
            try {
                // 删除旧文件缓存信息
                this.delSyncTaskCache(fileId);
                this.delBlockChecksumCache(fileId);
            } catch (Exception e) {
                log.error("删除文件同步任务等信息缓存失败", e);
            }
            // 保存新文件信息
            ResourceFileStorageBO destFileInfo = new ResourceFileStorageBO();
            // new params
            destFileInfo.setPid(newFileId);
            destFileInfo.setStatus(ResourceStatusEnum.NORMAL.getCode());
            destFileInfo.setIntegrity(FileIntegrityEnum.NORMAL.getCode());
            destFileInfo.setLength(FileUtil.size(destFile));
            destFileInfo.setStrongChecksum(strongChecksum);
            destFileInfo.setCatalog(storageConfig.getPersonalSpaceCatalog());
            destFileInfo.setResourceKey(newFileId);
            destFileInfo.setIsLocal(1);
            destFileInfo.setIsCloud(0);
            // old params
            destFileInfo.setVid(srcFileInfo.getVid());
            destFileInfo.setSpaceId(srcFileInfo.getSpaceId());
            destFileInfo.setSupId(srcFileInfo.getSupId());
            destFileInfo.setName(srcFileInfo.getName());
            destFileInfo.setSuffix(srcFileInfo.getSuffix());
            fileService.saveFile(destFileInfo, ConflictModeEnum.COVER_TAR_PURE);
            transactionManager.commit(status);
        } catch (Exception e) {
            // 事务回滚
            transactionManager.rollback(status);
            // 原文件状态回滚
            fileService.updateFileIntegrity(srcFileInfo.getPid(), originFileIntegrity);
            // 删除可能存在的新文件
            StorageHubUtil.silentDelete(destFile);
            throw DefaultSaasException.build("合并失败", e);
        }
        // 处理存储
        fileService.dealStorageMode(newFileId, clientDiffBlocks.getStorageMode(), true);
        return MapperUtil.nf().map(srcFileInfo, ResourceFileBO.class);
    }

    @Override
    public ServerDiffFileBO listServerDiffBlocks(ClientOriginFileBO clientOriginBlocks) {
        Assert.notNull(clientOriginBlocks, "客户端原数据为null");
        Assert.notNull(clientOriginBlocks.getBlockLength(), "block length is null");
        // 最新文件信息
        ResourceFileBO latestFile = fileService.getFile(clientOriginBlocks.getFileId(),
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
        this.checkFileIntegrity(latestFile);
        // 最新文件
        File fileData = fileService.getFileData(latestFile.getPid(), AssertEnum.FOUND);
        DiffBlockChecker checker = new DiffBlockChecker(fileData,
            MapperUtil.nf().mapAsList(clientOriginBlocks.getOriginBlocks(), OriginBlock.class), clientOriginBlocks.getBlockLength());
        checker.check();
        List<DiffBlock> diffCheckItems = checker.getDiffCheckItems();
        ServerDiffFileBO result = new ServerDiffFileBO();
        result.setFileId(latestFile.getPid());
        result.setVid(latestFile.getVid());
        result.setStrongChecksum(latestFile.getStrongChecksum());
        result.setDiffBlocks(MapperUtil.nf().mapAsList(diffCheckItems, DiffBlockBO.class));
        return result;
    }

    @Override
    public void saveClientDiffBlockData(String syncId, String blockId, InputStream ins) {
        log.debug("保存同步块({syncId: {}, blockId: {}})...", syncId, blockId);
        // check file status
        RsyncTaskBO syncTaskCache = this.getSyncTaskCache(syncId);
        if (Objects.isNull(syncTaskCache)) {
            throw new UnavailableException("同步任务已失效或不存在");
        }
        String fileId = syncTaskCache.getFileId();
        ResourceFileStorageBO fileInfo = fileService.getFileWithStorage(fileId,
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
        this.checkFileIntegrity(fileInfo);
        // save
        String supPath = catalogService.getCacheSupPath(storageConfig.getPersonalSpaceCacheCatalog(), syncId);
        chunkStorageService.saveChunk(supPath, blockId, ins);
    }

    @Override
    public void getServerDiffBlockData(String fileId, Integer blockPos, Integer blockLength, OutputStream ous) {
        // check file status
        log.debug("下载同步块({fileId: {}, blockPos: {}, blockLength: {}})...", fileId, blockPos, blockLength);
        ResourceFileStorageBO fileInfo = fileService.getFileWithStorage(fileId,
            LogicQuery.eq(ResourceStatusEnum.NORMAL.getCode()), LogicQuery.none(), AssertEnum.FOUND);
        this.checkFileIntegrity(fileInfo);
        File srcFile = fileService.getFileData(fileId, AssertEnum.FOUND);
        try {
            RsyncFileUtil.readBlock(srcFile, blockPos, blockLength, ous);
        } catch (IOException e) {
            throw DefaultSaasException.build("下载失败", e);
        }
    }

    private RsyncTaskBO addSyncTaskCache(String fileId, Integer blockLength, Integer blockStep) {
        RsyncTaskBO task = new RsyncTaskBO();
        task.setSyncId(StorageUtil.generateSyncTaskCacheKey(fileId));
        task.setFileId(fileId);
        task.setBlockLength(blockLength);
        task.setBlockStep(blockStep);
        RedisUtil.set(task.getSyncId(), task, StorageCacheConsts.RSYNC_TASK_CACHE_TIME);
        return task;
    }

    private RsyncTaskBO getSyncTaskCache(String syncId) {
        if (RedisUtil.hasKey(syncId)) {
            RedisUtil.expire(syncId, StorageCacheConsts.RSYNC_TASK_CACHE_TIME);
            return RedisUtil.get(syncId);
        }
        return null;
    }

    private void delSyncTaskCache(String fileId) {
        String keyPattern = StorageUtil.getSyncTaskCacheGroupKey(fileId).concat(CacheConsts.CACHE_KEY_WILDCARD);
        RedisUtil.delByPattern(keyPattern);
    }

    private void asyncAddBlockChecksumCache(String fileId, OriginBlockBuilder builder) {
        ThreadUtil.execute(() -> {
            // 异步存储校验和缓存
            List<OriginBlock> originBlocks = builder.getBlockChecksums();
            Integer checksumMode = null;
            if (CollUtil.isEmpty(originBlocks)) {
                // 未计算时，全部计算
                checksumMode = ChecksumModeEnum.ALL_CHECKSUM.code();
            } else {
                OriginBlock originBlock0 = originBlocks.get(0);
                if (Objects.isNull(originBlock0.getWeakChecksum()) && StrUtil.isBlank(originBlock0.getStrongChecksum())) {
                    checksumMode = ChecksumModeEnum.ALL_CHECKSUM.code();
                } else if (StrUtil.isBlank(originBlock0.getStrongChecksum())) {
                    checksumMode = ChecksumModeEnum.STRONG_CHECKSUM.code();
                } else if (Objects.isNull(originBlock0.getWeakChecksum())) {
                    checksumMode = ChecksumModeEnum.WEAK_CHECKSUM.code();
                }
            }
            if (Objects.nonNull(checksumMode)) {
                List<OriginBlock> anotherBlocks = builder.build(checksumMode);
                if (CollectionUtil.isNotEmpty(originBlocks)) {
                    if (!Objects.equals(anotherBlocks.size(), originBlocks.size())) {
                        log.error("强弱校验和计算分片数量不一致, fileId: {}, blockLength: {}, blockStep: {}, size: {} <-> {}",
                            fileId, builder.getBlockLength(), builder.getBlockStep(), originBlocks.size(), anotherBlocks.size());
                        return;
                    }
                    for (int i = 0; i < originBlocks.size(); i++) {
                        OriginBlock block1 = originBlocks.get(i);
                        OriginBlock block2 = anotherBlocks.get(i);
                        if (!Objects.equals(block1.getBlockLength(), block2.getBlockLength())
                            || !Objects.equals(block1.getBlockPos(), block2.getBlockPos())) {
                            log.error("强弱校验和计算分片列表中，同索引位置分片信息不一致，fileId: {}, blockLength: {}, blockStep: {}",
                                fileId, builder.getBlockLength(), builder.getBlockStep());
                            return;
                        }
                        if (Objects.nonNull(block2.getWeakChecksum()))
                            block1.setWeakChecksum(block2.getWeakChecksum());
                        if (StrUtil.isNotBlank(block2.getStrongChecksum()))
                            block1.setStrongChecksum(block2.getStrongChecksum());
                    }
                } else {
                    originBlocks = anotherBlocks;
                }
            }
            List<OriginBlockBO> blockList = MapperUtil.nf().mapAsList(originBlocks, OriginBlockBO.class);
            this.addBlockChecksumCache(fileId, builder.getBlockLength(), builder.getBlockStep(), blockList);
        });
    }

    private void addBlockChecksumCache(String fileId, Integer blockLength, Integer blockStep, List<OriginBlockBO> blockList) {
        String key = StorageUtil.getBLockChecksumCacheKey(fileId, blockLength, blockStep);
        RedisUtil.set(key, blockList, StorageCacheConsts.RSYNC_CHECKSUM_CACHE_TIME);
    }

    private List<OriginBlockBO> getBlockChecksumCache(String fileId, Integer blockLength, Integer blockStep) {
        String key = StorageUtil.getBLockChecksumCacheKey(fileId, blockLength, blockStep);
        if (RedisUtil.hasKey(key)) {
            RedisUtil.expire(key, StorageCacheConsts.RSYNC_CHECKSUM_CACHE_TIME);
            return RedisUtil.get(key);
        }
        return null;
    }

    private void delBlockChecksumCache(String fileId) {
        String keyPattern = StorageUtil.getBlockChecksumCacheGroupKey(fileId).concat(CacheConsts.CACHE_KEY_WILDCARD);
        RedisUtil.delByPattern(keyPattern);
    }

    private void checkFileIntegrity(ResourceFileBO fileInfo) {
        AssertUtil.assertThrow("文件正在合并中", HttpStatus.LOCKED, !Objects.equals(fileInfo.getIntegrity(), FileIntegrityEnum.NORMAL.getCode()));
    }

    private void assertChecksum(String srcStrongChecksum, String tarStrongChecksum) {
        boolean checksumError = StrUtil.isNotBlank(srcStrongChecksum) && !Objects.equals(srcStrongChecksum, tarStrongChecksum);
        AssertUtil.assertThrow(String.format("文件强校验和校验失败, expected: %s, actual: %s", srcStrongChecksum, tarStrongChecksum), checksumError);
        log.debug("rsync合并文件校验和比对通过.");
    }
}
